package com.education.realtime.util;


import com.education.realtime.annotation.NotSink;
import com.education.realtime.bean.TableProcess;
import com.education.realtime.bean.TradeProvinceOrderWindow;
import com.education.realtime.common.Constant;
import com.education.realtime.sink.PhoenixSink;

import com.alibaba.fastjson.JSONObject;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.calcite.shaded.com.google.common.base.CaseFormat;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;
import java.util.stream.Collectors;
import java.util.stream.Stream;


public class FlinkSinkUtil {
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
    }
    
    public static SinkFunction<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);


        return new FlinkKafkaProducer<String>(
                "default",
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                    @Nullable Long timestamp) {
                        return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }


    public static <T> SinkFunction<T> getJdbcSink(String driver,
                                                  String url,
                                                  String sql,
                                                  String user,
                                                  String password){

        return JdbcSink.sink(sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps,
                                       T t) throws SQLException {
                        // 给ps中的占位符进行赋值
                        // 从t对象中, 取出值, 给sql中的占位符赋值
                        // insert into t(a,b,c,d)values(?,?,?,?)
                        // 获取Class几种方式:
                        // 1. 类名.class 2. Class.forName("...") 3. 对象.getClass
                        Field[] fields = t.getClass().getDeclaredFields();

                        try {
                            for (int i = 0; i < fields.length; i++) {
                                Field field = fields[i];
                                field.setAccessible(true);
                                ps.setObject(i + 1, field.get(t));
                            }
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchIntervalMs(1000)
                        .withBatchSize(1024)
                        .withMaxRetries(3)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(driver)
                        .withUrl(url)
                        .withUsername(user)
                        .withPassword(password)
                        .build()
        );
    }

    public static <T> SinkFunction<T> getClickHouseSink(String db,
                                                        String table,
                                                        Class<T> tClass){

        String driver = Constant.CLICKHOUSE_DRIVER;
        String url = Constant.CLICKHOUSE_URL_PER + db;


        //TODO clickhouse插入语句 insert into tablename(a,b,c) values(?,?,?)
        //获取一个对象数组，表示该类或接口声明的所有字段 。这包括公共、受保护、默认 *（包）访问和私有字段，但不包括继承的字段。
        Field[] fields = tClass.getDeclaredFields();
        String columns = Stream.of(fields)//将fields数组转换成一个流，流式编程
                .filter(field -> field.getAnnotation(NotSink.class) == null)
                .map(f -> {
                    String name = f.getName();
                    return CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, name);//驼峰转下划线形式

                })
                .collect(Collectors.joining(","));//按指定的分割符分割收集元素

        StringBuilder sql = new StringBuilder();
        sql
                .append("insert into ")
                .append(table)
                .append("(")
                .append(columns)
                .append(")values(")
                .append(columns.replaceAll("[^,]+","?"))
                .append(")");

        System.out.println("clickhouse插入语句: " + sql.toString());

        return getJdbcSink(driver,url,sql.toString(),null,null);


    }
}
